package com.znz.analysis.spark.sparkContext;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.apache.spark.SparkConf;
import org.springframework.stereotype.Component;

import com.znz.analysis.constant.properties.SparkProperties;

/**
 * 创建SparkConf并初始化
 * 
 * @author 作者 :TP
 * @version 创建时间：2017年6月29日 上午10:21:42
 */

@Component
public class SparkConfFactory {

	@Resource
	private SparkConf sparkConf;

	@Resource
	private SparkProperties sparkProperties;

	public SparkConf getInstance() {
		return this.sparkConf;
	}

	/**
	 * 初始化创建SparkConf对象
	 */
	@PostConstruct
	private void init() {
		System.out.println("PostConstruct: Init SparkConf");
		sparkConf.setAppName(sparkProperties.getApp_name()).setMaster(sparkProperties.getMaster_url())
				.set("spark.executor.memory", sparkProperties.getExecutor_memory())
				// 开启map端输出文件的合并机制
				.set("spark.shuffle.consolidateFiles", "true")
				// 设置map端内存缓冲区大小（默认32k）
				.set("spark.shuffle.file.buffer", "64k")
				// 设置reduce端内存占比（默认内存比例是0.2）
				.set("spark.shuffle.memoryFraction", "0.4").setJars(getAssemblyJars());
		System.out.println("Init SparkConf Success");
	}

	/**
	 * 获取系统目录下的jar包路径
	 * 
	 * @return
	 */
	private String[] getAssemblyJars() {
		String baseDir = sparkProperties.getLib_dir();
		String[] jars = new String[] { baseDir + "/spark-streaming-kafka_2.10-1.4.1.jar",
				baseDir + "/kafka_2.10-0.8.2.1.jar", baseDir + "/zkclient-0.3.jar", baseDir + "/metrics-core-2.2.0.jar",
				baseDir + "/kafka-clients-0.8.2.1.jar", baseDir + "/fastjson-1.2.16.jar",
				baseDir + "/json-lib-2.4-jdk15.jar", baseDir + "/ezmorph-1.0.6.jar", baseDir + "/geoip2-2.1.0.jar",
				baseDir + "/maxmind-db-1.0.0.jar", baseDir + "/DataAnalysis-1.0.0.RC1.jar" };
		return jars;
	}

}
